package org.zjt.book

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by yp-pd-7033 on 2017/6/28.
  */
object sparkMysql {

  case class Blog(name: String, password: String, age: Int)


  def myFun(colums: Iterable[String], data: RDD[Map[String, Any]]): Unit = {

    var sql = "insert into t_result("
    colums.toArray.sorted.foreach {
      y =>
        sql = sql + y + ","
    }
    sql = sql.substring(0, sql.length - 1) + ") values"
    println("NEW SQL:" + sql)

    var avalue = ""
    val sql2 = data.map {
      x => {
        val s = colums.toArray.sorted.map(x.getOrElse(_, null)).mkString(",")
        if (s.isEmpty) s else s"($s)"
      }
    }
    println(sql2.collect().mkString(","))


    data.foreach {
      x => {
        var bvalue = "("
        colums.foreach {
          y => bvalue = bvalue + x(y) + ","
        }
        bvalue = bvalue.substring(0, bvalue.length - 1) + "),"
        avalue = avalue + bvalue
        println("--" + avalue)
      }
    }

    println("--:" + avalue)
    avalue = avalue.substring(0, avalue.length - 1)
    println(avalue + "---" + avalue.length)
    sql = sql + avalue
    println("NEW SQL:" + sql)


  }

  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("RDDToMysql").setMaster("local")
    val sc = new SparkContext(conf)
    val map1 = Map("name" -> "张三", "age" -> 12, "address" -> "ssssd")
    val map2 = Map("name" -> "王问", "address" -> "www.3", "age" -> 30)
    val map3 = Map("name" -> "李四", "address" -> "www.2", "age" -> 20)
    val data = sc.parallelize(List(map1, map2, map3))

    val colums = map1.keys
    //    data.foreachPartition(myFun)
    myFun(colums, data)
  }


  def myFun(iterator: Iterator[(String, String, Int)]): Unit = {
    var conn: Connection = null
    var ps: PreparedStatement = null
    val sql = "insert into t_result(name, password, age) values (?, ?, ?)"
    try {
      conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/Test", "root", "123456")
      iterator.foreach(data => {
        ps = conn.prepareStatement(sql)
        ps.setString(1, data._1)
        ps.setString(2, data._2)
        ps.setInt(3, data._3)
        ps.executeUpdate()
      }
      )
    } catch {
      case e: Exception => println("Mysql Exception")
    } finally {
      if (ps != null) {
        ps.close()
      }
      if (conn != null) {
        conn.close()
      }
    }
  }
}
